package com.atguigu.utils;

import com.atguigu.annotation.TransientSink;
import com.atguigu.common.DWSStaticConstants;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class ClickHouseUtil {
    public static <T> SinkFunction<T> getSink(String sql) {

        return JdbcSink.<T>sink(sql, new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        Field[] fields = t.getClass().getDeclaredFields();
                        int offset = 0;
                        for (int i = 0; i < fields.length; i++) {
                            try {
                                Object value = fields[i].get(t);
                                fields[i].setAccessible(true);
                                TransientSink annotation = fields[i].getAnnotation(TransientSink.class);
                                if (annotation != null){
                                    offset++;
                                    continue;
                                }
                                    preparedStatement.setObject(i + 1 - offset, value);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                new JdbcExecutionOptions
                        .Builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions
                        .JdbcConnectionOptionsBuilder()
                        .withDriverName(DWSStaticConstants.CLICKHOUSE_DRIVER)
                        .withUrl(DWSStaticConstants.CLICKHOUSE_URL).build());
    }
}
